package iot.hbase;

import hbase.template.HBaseTemplate;
import iot.entity.Location;
import iot.hbase.codec.LocationCodec;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.List;

/**
 * Created by dave on 18-7-21 下午11:39.
 */
@Service
@ConditionalOnExpression("${hbase.async:false}")
@ConditionalOnProperty(name = "dao.type", havingValue = "hbase")
public class LocationDao {
    private static final String TABLE = "location";
    @Autowired HBaseTemplate template;
    private LocationCodec codec = new LocationCodec();

    public void insert(Location location) {
        template.put(TABLE, location, codec);
    }

    public void batchInsert(List<Location> locations) {
        template.batchInsert(TABLE, locations, codec);
    }

    public List<Location> getList(String sim, Date start, Date end, int limit, boolean reversed) {
        Scan scan = new Scan();
        scan.withStartRow(codec.makeStartRowKey(sim, start));
        scan.setFilter(new PageFilter(limit));
        scan.addFamily("l".getBytes());
        if (reversed) {
            scan.setReversed(true);
            scan.withStopRow(codec.makeEndRowKey(sim, end), true);
        } else {
            scan.withStopRow(codec.makeEndRowKey(sim, end));
        }
        return template.find(TABLE, scan, codec);
    }

    public List<Location> getListAfter(String sim, Date start, int limit) {
        Scan scan = new Scan();
        scan.withStartRow(codec.makeStartRowKey(sim, start));
        scan.setFilter(new PageFilter(limit));
        scan.addFamily("l".getBytes());
        return template.find(TABLE, scan, codec);
    }

    public List<Location> getListBefore(String sim, Date end, int limit) {
        Scan scan = new Scan();
        scan.setReversed(true);
        scan.addFamily("l".getBytes());
        scan.setFilter(new PageFilter(limit));
        scan.withStartRow(codec.makeEndRowKey(sim, end), true);
        return template.find(TABLE, scan, codec);
    }

    public long rowCount(String sim, Date start, Date end) {
        Scan scan = new Scan();
        scan.withStartRow(codec.makeStartRowKey(sim, start));
        scan.withStopRow(codec.makeEndRowKey(sim, end));
        scan.setFilter(new FirstKeyOnlyFilter());
        return template.rowCount(TABLE, scan);
    }

    public long rowCount() {
        Scan scan = new Scan();
        scan.setFilter(new FirstKeyOnlyFilter());
        return template.rowCount(TABLE, scan);
    }
}
